package com.sh.data.engine.domain.datadev.flink.service.impl;

import cn.hutool.core.convert.Convert;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.sh.data.engine.common.enumDefinition.FlinkNodeTypeEnum;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.datadev.flink.model.domain.*;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeConfigService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeService;
import com.sh.data.engine.domain.workspace.manager.model.domain.UserDomain;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.repository.dao.datadev.FlinkNodeConfigMapper;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkMenuNodeEntity;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkNodeConfigEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.File;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author: zigui.zdf
 * @description:
 * @date: 2021/6/3 10:49
 */
@Service
@Slf4j
public class FlinkNodeConfigServiceImpl
    extends ServiceImpl<FlinkNodeConfigMapper, FlinkNodeConfigEntity>
    implements FlinkNodeConfigService {

    @Value("${flink.resource.path:/data/}")
    private String flinkResourcePath;

    @Lazy
    @Autowired
    private FlinkNodeService flinkNodeService;

    @Autowired
    private UserService userService;

    @Override
    public UploadFlinkResourceDomain uploadFlinkResource(
        Long projectId, String fileName, byte[] data) {
        String newFileName = UUID.randomUUID().toString();
        newFileName = String.format("%s.jar", newFileName);
        String path =
            String.format(
                "%s%s%s%s%s",
                flinkResourcePath, File.separator, projectId, File.separator, newFileName);
        String userId = UserContext.getUserId();

        FlinkNodeConfigEntity flinkResourceEntity = new FlinkNodeConfigEntity();
        File file = new File(path);
        try {
            FileUtils.writeByteArrayToFile(file, data);
        } catch (IOException e) {
            throw new BusinessException("上传失败");
        }
        Date date = new Date();
        FlinkNodeConfigEntity.ResourceConfig resourceConfig =
            new FlinkNodeConfigEntity.ResourceConfig();
        resourceConfig.setFileName(fileName);
        resourceConfig.setPath(path);
        flinkResourceEntity.setResourceConfig(resourceConfig);
        flinkResourceEntity.setCreatorId(userId);
        flinkResourceEntity.setCreateTime(date);
        flinkResourceEntity.setProjectId(projectId);
        flinkResourceEntity.setUpdaterId(userId);
        flinkResourceEntity.setUpdateTime(date);

        this.save(flinkResourceEntity);
        UploadFlinkResourceDomain flinkResourceDomain =
            UploadFlinkResourceDomain.builder()
                .resourcrId(flinkResourceEntity.getId())
                .filePath(path)
                .build();
        return flinkResourceDomain;
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public boolean saveFlinkResource(FlinkNodeConfigResourceDomain saveResourceParam) {
        String userId = UserContext.getUserId();
        Long projectId = ServletUtils.getProjectId();
        //    FlinkNodeConfigEntity flinkNodeConfigEntity =
        //        flinkNodeConfigMapper.findById(saveResourceParam.getNodeId());

        //    Criteria criteria =
        //        Criteria.where("nodeId")
        //            .is(nodeId)
        //            //            .and("nodeType")
        //            //            .is(FlinkNodeTypeEnum.RESOURCE.getNodeType())
        //            .and("active")
        //            .is(true);
        //    query.addCriteria(criteria).limit(1);

        LambdaQueryWrapper<FlinkNodeConfigEntity> last =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getNodeId, saveResourceParam.getNodeId())
                .eq(FlinkNodeConfigEntity::getIsActive, 1)
                .last("limit 1");
        FlinkNodeConfigEntity flinkNodeConfigEntity = getOne(last);
        if (Objects.isNull(flinkNodeConfigEntity)) {
            flinkNodeConfigEntity = new FlinkNodeConfigEntity();
        }
        boolean check =
            flinkNodeService.hasSameMenuNode(
                saveResourceParam.getNodeId(),
                saveResourceParam.getPid(),
                projectId,
                saveResourceParam.getResourceName(),
                FlinkNodeTypeEnum.RESOURCE.getCode());
        if (check) {
            throw new BusinessException("存在同名资源");
        }
        /** 不存在nodeId 新建，存在替换 */
        if (Objects.isNull(saveResourceParam.getNodeId())) {
            FlinkMenuNodeEntity flinkMenuNodeEntity = new FlinkMenuNodeEntity();
            Date date = new Date();
            flinkMenuNodeEntity.setCreatorId(userId);
            flinkMenuNodeEntity.setCreateTime(date);
            flinkMenuNodeEntity.setUpdaterId(userId);
            flinkMenuNodeEntity.setUpdateTime(date);
            //      flinkMenuNodeEntity.setIsDir(0);
            flinkMenuNodeEntity.setIsLocked(0);
            flinkMenuNodeEntity.setIsSystem(0);
            flinkMenuNodeEntity.setPid(saveResourceParam.getPid());
            flinkMenuNodeEntity.setName(saveResourceParam.getResourceName());
            flinkMenuNodeEntity.setProjectId(projectId);
            flinkMenuNodeEntity.setNodeType(FlinkNodeTypeEnum.RESOURCE.getCode());
            flinkMenuNodeEntity.setRemark(saveResourceParam.getRemark());
            flinkNodeService.save(flinkMenuNodeEntity);
            flinkNodeConfigEntity.setNodeId(flinkMenuNodeEntity.getId());
            flinkNodeConfigEntity.setProjectId(projectId);
            flinkNodeConfigEntity.setIsActive(1);
            FlinkNodeConfigEntity.ResourceConfig resourceConfig =
                new FlinkNodeConfigEntity.ResourceConfig();
            resourceConfig.setIsActive(1);
            resourceConfig.setFileName(saveResourceParam.getOriginalFilename());
            resourceConfig.setFileDownloadUrl(saveResourceParam.getFileDownloadUrl());
            resourceConfig.setResourceName(saveResourceParam.getResourceName());
            resourceConfig.setRemark(saveResourceParam.getRemark());
            resourceConfig.setPath(saveResourceParam.getFilePath());
            flinkNodeConfigEntity.setResourceConfig(resourceConfig);
            flinkNodeConfigEntity.setCreatorId(userId);
            flinkNodeConfigEntity.setNodeType(FlinkNodeTypeEnum.RESOURCE.getCode());
            flinkNodeConfigEntity.setId(null);
            this.save(flinkNodeConfigEntity);
        } else {
            // 更新节点信息
            FlinkMenuNodeEntity update = new FlinkMenuNodeEntity();
            update.setId(saveResourceParam.getNodeId());
            update.setPid(saveResourceParam.getPid());
            update.setRemark(saveResourceParam.getRemark());
            update.setName(saveResourceParam.getResourceName());
            flinkNodeService.updateById(update);
            // 获取之前节点配置 版本更新为过期
            if (StringUtils.isNotBlank(saveResourceParam.getResourceName())
                && StringUtils.isNotBlank(saveResourceParam.getFilePath())
                && StringUtils.isNotBlank(saveResourceParam.getOriginalFilename())) {
                FlinkNodeConfigEntity resourceNodeConfig =
                    this.getActiveResourceByNodeId(saveResourceParam.getNodeId());
                resourceNodeConfig.setUpdateTime(new Date());
                resourceNodeConfig.setUpdaterId(userId);
                resourceNodeConfig.setIsActive(0);
                resourceNodeConfig.getResourceConfig().setIsActive(0);
                this.updateById(resourceNodeConfig);
                // 插入新的resource配置
                flinkNodeConfigEntity.setProjectId(projectId);
                flinkNodeConfigEntity.setCreatorId(userId);
                flinkNodeConfigEntity.setUpdaterId(userId);
                flinkNodeConfigEntity.setCreateTime(new Date());
                flinkNodeConfigEntity.setUpdateTime(new Date());
                flinkNodeConfigEntity.setIsActive(1);
                flinkNodeConfigEntity.setNodeId(saveResourceParam.getNodeId());
                FlinkNodeConfigEntity.ResourceConfig resourceConfig =
                    new FlinkNodeConfigEntity.ResourceConfig();
                resourceConfig.setIsActive(1);
                resourceConfig.setFileName(saveResourceParam.getOriginalFilename());
                resourceConfig.setRemark(saveResourceParam.getRemark());
                resourceConfig.setFileDownloadUrl(saveResourceParam.getFileDownloadUrl());
                resourceConfig.setPath(saveResourceParam.getFilePath());
                resourceConfig.setResourceName(saveResourceParam.getResourceName());
                flinkNodeConfigEntity.setResourceConfig(resourceConfig);
                flinkNodeConfigEntity.setId(null);
                this.save(flinkNodeConfigEntity);
            }
        }

        return true;
    }

    @Override
    public FlinkNodeConfigEntity getActiveResourceByNodeId(Long nodeId) {
        LambdaQueryWrapper<FlinkNodeConfigEntity> last =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getNodeId, nodeId)
                .eq(FlinkNodeConfigEntity::getIsActive, 1)
                .last("limit 1");
        FlinkNodeConfigEntity one = this.getOne(last);
        return one;
    }

    @Override
    public List<FlinkNodeConfigEntity> getRunnableTaskList(FlinkNodeTypeEnum flinkNodeTypeEnum) {

        LambdaQueryWrapper<FlinkNodeConfigEntity> and =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getNodeType, flinkNodeTypeEnum.getCode())
                .eq(FlinkNodeConfigEntity::getIsPublished, 1)
                .eq(FlinkNodeConfigEntity::getIsActive, 1)
                .and(
                    e ->
                        e.lt(FlinkNodeConfigEntity::getNextExecTime, new Date())
                            .or()
                            .gt(FlinkNodeConfigEntity::getNextStopTime, new Date()));

        List<FlinkNodeConfigEntity> list = list(and);
        return list;
    }

    @Override
    public List<FlinkNodeConfigEntity> getAllTaskList(FlinkNodeTypeEnum flinkNodeTypeEnum) {

        LambdaQueryWrapper<FlinkNodeConfigEntity> eq =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getNodeType, flinkNodeTypeEnum.getCode())
                .eq(FlinkNodeConfigEntity::getIsPublished, 1)
                .eq(FlinkNodeConfigEntity::getIsActive, 1);
        List<FlinkNodeConfigEntity> list = list(eq);

        return list;
    }

    @Override
    public List<FlinkNodeConfigEntity> getStopedTaskList(FlinkNodeTypeEnum flinkNodeTypeEnum) {
        //  public List<FlinkNodeConfigEntity> getStopedTaskList(FlinkNodeTypeEnum flinkNodeTypeEnum) {
        //    Query query = new Query();
        //    query.addCriteria(
        //        Criteria.where("nodeType")
        //            .is(flinkNodeTypeEnum.getCode())
        //            .and("published")
        //            .is(true)
        //            .and("active")
        //            .is(true)
        //            .and("nextStopTime")
        //            .lte(new Date()));
        //    return findMulti(query);
        //  }

        LambdaQueryWrapper<FlinkNodeConfigEntity> lt =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getNodeType, flinkNodeTypeEnum.getCode())
                .eq(FlinkNodeConfigEntity::getIsPublished, 1)
                .eq(FlinkNodeConfigEntity::getIsActive, 1)
                .lt(FlinkNodeConfigEntity::getNextStopTime, new Date());
        List<FlinkNodeConfigEntity> list = list(lt);
        return list;
    }

    @Override
    public IPage<FlinkNodeConfigEntity> getVersionsTaskList(
        FlinkNodeTypeEnum taskSql, Long nodeId, Integer pageNum, Integer pageSize) {

        //    Query query = new Query();
        //    query.addCriteria(
        //        Criteria.where("published")
        //            .is(true)
        //            .and("active")
        //            .is(false)
        //            .and("nodeId")
        //            .is(nodeId));
        //    query.with(Sort.by(Sort.Direction.DESC, "publishTime"));

        LambdaQueryWrapper<FlinkNodeConfigEntity> eq =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getIsPublished, 1)
                .eq(FlinkNodeConfigEntity::getIsActive, 0)
                .eq(FlinkNodeConfigEntity::getNodeId, nodeId)
                .orderByDesc(FlinkNodeConfigEntity::getPublishTime);

        Page<FlinkNodeConfigEntity> page = page(new Page<>(pageNum, pageSize), eq);
        return page;
    }

    @Override
    public void rollbackVersion(FlinkNodeTypeEnum flinkNodeTypeEnum, Long nodeId, Long versionId) {
        //    Query query = new Query();
        //    query.addCriteria(
        //        Criteria.where("nodeId")
        //            .is(nodeId)
        //            .and("active")
        //            .is(true));
        //    Update update = new Update();
        //    update.set("active", false);
        //    update.set("updateTime", new Date());
        //    update(query, update);
        //    FlinkNodeConfigEntity entity = new FlinkNodeConfigEntity();
        //    entity.setId(versionId);
        //    entity.setActive(true);
        //    entity.setUpdateTime(new Date());
        //    this.updateByIdSelective(entity);
        UpdateWrapper<FlinkNodeConfigEntity> wrapper = Wrappers.update();
        wrapper
            .lambda()
            .eq(FlinkNodeConfigEntity::getNodeId, nodeId)
            .eq(FlinkNodeConfigEntity::getIsActive, 1);

        FlinkNodeConfigEntity flinkNodeConfigEntity = new FlinkNodeConfigEntity();
        flinkNodeConfigEntity.setIsActive(0);
        flinkNodeConfigEntity.setUpdateTime(new Date());
        flinkNodeConfigEntity.setId(versionId);
        this.update(flinkNodeConfigEntity, wrapper);
    }

    @Override
    public List<FlinkNodeConfigDomain> getNodeConfigByNodeId(Long nodeId) {

        LambdaQueryWrapper<FlinkNodeConfigEntity> eq =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getNodeId, nodeId);
        List<FlinkNodeConfigEntity> list = list(eq);
        return Convert.toList(FlinkNodeConfigDomain.class, list);
    }

    @Override
    public FlinkNodeConfigDomain getNodeConfigById(Long id) {
        FlinkNodeConfigEntity nodeConfig = this.getById(id);
        return Convert.convert(FlinkNodeConfigDomain.class, nodeConfig);
    }

    @Override
    public FlinkNodeConfigResourceDomain getActiveFlinkResource(Long nodeId) {
        FlinkNodeConfigEntity nodeConfigEntity = this.getActiveResourceByNodeId(nodeId);
        FlinkNodeConfigResourceDomain resourceConfigDomain =
            Convert.convert(FlinkNodeConfigResourceDomain.class, nodeConfigEntity);
        FlinkNodeConfigEntity.ResourceConfig resourceConfig = nodeConfigEntity.getResourceConfig();
        if (!Objects.isNull(resourceConfig)) {
            FlinkMenuNodeEntity one = flinkNodeService.getById(nodeId);
            resourceConfigDomain.setResourceName(one.getName());
            resourceConfigDomain.setFilePath(resourceConfig.getPath());
            resourceConfigDomain.setRemark(one.getRemark());
            resourceConfigDomain.setFileDownloadUrl(resourceConfig.getFileDownloadUrl());
        }
        UserDomain userDomain = userService.getUserById(nodeConfigEntity.getCreatorId());
        resourceConfigDomain.setUserName(
            Optional.ofNullable(userDomain).map(UserDomain::getUserName).orElse(""));
        return resourceConfigDomain;
    }

    @Override
    public IPage<FlinkNodeConfigResourceDomain> getFlinkResourcePage(
        Long nodeId, Integer pageNum, Integer pageSize) {

        //    if (StringUtils.isNotBlank(nodeId)) {
        //      query.addCriteria(Criteria.where("nodeId").is(nodeId));
        //    }
        //    final Sort sort = Sort.by(Sort.Direction.DESC, "createTime");
        //    query.with(sort);
        //    MongoPageHelper pageHelper = new MongoPageHelper(mongoTemplate);
        //    PageResult<FlinkNodeConfigEntity> nodeConfigPage =
        //        pageHelper.pageQuery(query, FlinkNodeConfigEntity.class, pageSize, pageNum);

        LambdaQueryWrapper<FlinkNodeConfigEntity> lambdaQueryWrapper =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getNodeId, nodeId)
                .orderByDesc(FlinkNodeConfigEntity::getCreateTime);
        IPage<FlinkNodeConfigEntity> convert = page(new Page<>(pageNum, pageSize), lambdaQueryWrapper);

        List<FlinkNodeConfigEntity> result = convert.getRecords();

        IPage<FlinkNodeConfigResourceDomain> r1 =
            convert.convert(e -> Convert.convert(FlinkNodeConfigResourceDomain.class, e));
        if (CollectionUtils.isNotEmpty(result)) {
            List<String> userIds = result.stream().map(r -> r.getCreatorId()).collect(Collectors.toList());
            Map<String, String> userMap = userService.getMapByUserIds(userIds);
            List<FlinkNodeConfigResourceDomain> domain =
                result.stream()
                    .map(
                        r -> {
                            FlinkNodeConfigResourceDomain resourceConfigDomain =
                                Convert.convert(FlinkNodeConfigResourceDomain.class, r);
                            FlinkNodeConfigEntity.ResourceConfig resourceConfig = r.getResourceConfig();
                            if (!Objects.isNull(resourceConfig)) {
                                resourceConfigDomain.setResourceName(resourceConfig.getResourceName());
                                resourceConfigDomain.setFileName(resourceConfig.getFileName());
                                resourceConfigDomain.setFilePath(resourceConfig.getPath());
                                resourceConfigDomain.setRemark(resourceConfig.getRemark());
                                resourceConfigDomain.setFileDownloadUrl(resourceConfig.getFileDownloadUrl());
                            }
                            resourceConfigDomain.setUserName(userMap.get(r.getCreatorId()));
                            return resourceConfigDomain;
                        })
                    .collect(Collectors.toList());
            r1.setRecords(domain);
        }
        return r1;
    }

    @Override
    public void deleteResourceById(Long id) {
        FlinkNodeConfigEntity byId = this.getById(id);
        if (byId.getResourceConfig().getIsActive().equals(1)) {
            throw new BusinessException("最新的资源不能删除");
        }
        ;
        // TODO 判断资源是否被函数引用，否则不能删除
        this.removeById(id);
    }

    @Override
    public void editFlinkResourceByNodeId(Long nodeId, String resourceName) {
        String userId = UserContext.getUserId();

        FlinkMenuNodeEntity nodeEntity = flinkNodeService.getById(nodeId);
        nodeEntity.setName(resourceName);
        nodeEntity.setUpdateTime(new Date());
        flinkNodeService.updateById(nodeEntity);

        FlinkNodeConfigEntity flinkResourceEntity = this.getActiveResourceByNodeId(nodeId);
        flinkResourceEntity.setUpdaterId(userId);
        flinkResourceEntity.setUpdateTime(new Date());
        flinkResourceEntity.getResourceConfig().setFileName(resourceName);
        this.updateById(flinkResourceEntity);
    }

    @Override
    public void saveFunctionConfig(
        Long nodeId,
        Integer functionType,
        String className,
        Long resourceNodeId,
        String purpose,
        String format,
        String paramDesc,
        String userId,
        Long projectId) {

        List<FlinkNodeConfigEntity> nodeConfigs = this.getConfigByNodeId(nodeId);

        FlinkNodeConfigEntity flinkFunctionEntity = null;
        if (CollectionUtils.isNotEmpty(nodeConfigs)) {
            flinkFunctionEntity = nodeConfigs.get(0);
        }

        if (null == flinkFunctionEntity) {
            flinkFunctionEntity = new FlinkNodeConfigEntity();
            flinkFunctionEntity.setProjectId(projectId);
            flinkFunctionEntity.setNodeId(nodeId);

            FlinkNodeConfigEntity.FunctionConfig functionConfig =
                new FlinkNodeConfigEntity.FunctionConfig();
            functionConfig.setFunctionType(functionType);
            functionConfig.setClassName(className);
            functionConfig.setResourceNodeId(resourceNodeId);
            functionConfig.setPurpose(purpose);
            functionConfig.setFormat(format);
            functionConfig.setParamDesc(paramDesc);
            flinkFunctionEntity.setFunctionConfig(functionConfig);

            flinkFunctionEntity.setCreatorId(userId);
            flinkFunctionEntity.setUpdaterId(userId);
            flinkFunctionEntity.setPublisherId(userId);
            flinkFunctionEntity.setIsPublished(1);
            flinkFunctionEntity.setIsActive(1);
            Date date = new Date();
            flinkFunctionEntity.setPublishTime(date);
            flinkFunctionEntity.setCreateTime(date);
            flinkFunctionEntity.setUpdateTime(date);
            flinkFunctionEntity.setNodeType(FlinkNodeTypeEnum.FUNCTION.getCode());
            this.save(flinkFunctionEntity);
        } else {
            flinkFunctionEntity.getFunctionConfig().setClassName(className);
            flinkFunctionEntity.getFunctionConfig().setResourceNodeId(resourceNodeId);
            flinkFunctionEntity.getFunctionConfig().setPurpose(purpose);
            flinkFunctionEntity.getFunctionConfig().setFormat(format);
            flinkFunctionEntity.getFunctionConfig().setParamDesc(paramDesc);
            flinkFunctionEntity.setUpdaterId(userId);
            flinkFunctionEntity.setUpdateTime(new Date());
            this.updateById(flinkFunctionEntity);
        }
    }

    private List<FlinkNodeConfigEntity> getConfigByNodeId(Long nodeId) {

        LambdaQueryWrapper<FlinkNodeConfigEntity> eq =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getNodeId, nodeId);
        List<FlinkNodeConfigEntity> list = list(eq);
        return list;
    }

    @Override
    public void insertOne(FlinkNodeConfigEntity flinkNodeConfigEntity) {
        this.save(flinkNodeConfigEntity);
    }

    @Override
    public void updateOne(FlinkNodeConfigEntity flinkNodeConfigEntity) {
        this.updateById(flinkNodeConfigEntity);
    }

    @Override
    public FlinkNodeFunctionDomain getFunctionDetail(Long nodeId) {

        FlinkNodeDomain nodeDomain = flinkNodeService.getFinkNodeById(nodeId);

        //    final FlinkMenuNodeDomain nodeDomain = flinkMenuNodeService.getById(nodeId);
        if (Objects.isNull(nodeDomain)) {
            throw new BusinessException("数据不存在");
        }
        FlinkNodeFunctionDomain nodeFunctionDomain = new FlinkNodeFunctionDomain();
        nodeFunctionDomain.setNodeName(nodeDomain.getName());
        nodeFunctionDomain.setFunctionName(nodeDomain.getName());
        nodeFunctionDomain.setNodeType(nodeDomain.getNodeType());

        List<FlinkNodeConfigEntity> functionConfigEntities = this.getConfigByNodeId(nodeId);
        if (CollectionUtils.isEmpty(functionConfigEntities)) {
            throw new BusinessException("函数配置详情不存在，请检查后重试！");
        }
        final FlinkNodeConfigEntity functionDomain = functionConfigEntities.get(0);

        if (null != functionDomain) {
            nodeFunctionDomain.setFormat(functionDomain.getFunctionConfig().getFormat());
            nodeFunctionDomain.setPurpose(functionDomain.getFunctionConfig().getPurpose());
            nodeFunctionDomain.setParamDesc(functionDomain.getFunctionConfig().getParamDesc());
            nodeFunctionDomain.setClassName(functionDomain.getFunctionConfig().getClassName());
            nodeFunctionDomain.setResourceNodeId(functionDomain.getFunctionConfig().getResourceNodeId());
            FlinkNodeConfigEntity byId =
                this.getById(functionDomain.getFunctionConfig().getResourceNodeId());
            FlinkMenuNodeEntity nodeInfo = flinkNodeService.getById(byId.getNodeId());
            String resourceName = nodeInfo.getName();
            nodeFunctionDomain.setResourceName(resourceName);
            nodeFunctionDomain.setCreateTime(functionDomain.getCreateTime());
            nodeFunctionDomain.setUpdateTime(functionDomain.getUpdateTime());
            UserDomain user = userService.getUserById(functionDomain.getCreatorId());
            if (!Objects.isNull(user)) {
                nodeFunctionDomain.setUsername(user.getUserName());
            }
            if (Objects.equals(functionDomain.getCreatorId(), functionDomain.getUpdaterId())) {
                nodeFunctionDomain.setModifyUsername(nodeFunctionDomain.getUsername());
            } else {
                UserDomain updateUser = userService.getUserById(functionDomain.getUpdaterId());
                if (!Objects.isNull(updateUser)) {
                    nodeFunctionDomain.setModifyUsername(updateUser.getUserName());
                }
            }
            if (null != functionDomain.getFunctionConfig().getResourceNodeId()) {
                //        FlinkNodeConfigResourceDomain resourceDomain =
                //
                // this.getActiveFlinkResource(functionDomain.getFunctionConfig().getResourceNodeId());
                FlinkNodeConfigEntity resourceDomain =
                    this.getById(functionDomain.getFunctionConfig().getResourceNodeId());
                if (null != resourceDomain) {
                    nodeFunctionDomain.setResourcePath(resourceDomain.getResourceConfig().getPath());
                }
            }
        }
        return nodeFunctionDomain;
    }

    @Override
    public FlinkNodeConfigEntity getActiveByNodeId(Long nodeId) {
        FlinkNodeConfigEntity activeResourceByNodeId = this.getActiveResourceByNodeId(nodeId);
        return activeResourceByNodeId;
    }

    @Override
    public void deleteByNodeId(Long nodeId) {
        LambdaQueryWrapper<FlinkNodeConfigEntity> eq =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getNodeId, nodeId);
        this.remove(eq);
    }

    @Override
    public List<FlinkNodeConfigEntity> getByNodeIds(List<Long> nodeIds) {

        LambdaQueryWrapper<FlinkNodeConfigEntity> in =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>().eq(FlinkNodeConfigEntity::getIsActive, 1);

        if (org.apache.commons.collections4.CollectionUtils.isNotEmpty(nodeIds)) {
            in.in(FlinkNodeConfigEntity::getNodeId, nodeIds);
        }
        return this.list(in);
    }

    @Override
    public Map<Long, Integer> getScheduleTypeMapByIds(List<Long> ids) {

        if (CollectionUtils.isEmpty(ids)) {
            return Collections.EMPTY_MAP;
        }
        List<FlinkNodeConfigEntity> multi = this.listByIds(ids);

        Map<Long, Integer> result =
            multi.stream()
                .collect(
                    Collectors.toMap(
                        i -> i.getId(),
                        i ->
                            Objects.isNull(i.getScheduleConfig())
                                ? 0
                                : i.getScheduleConfig().getEffectDayType()));

        return result;
    }

    @Override
    public List<FlinkNodeConfigEntity> getActiveResourceByProjectId(Long projectId) {

        LambdaQueryWrapper<FlinkNodeConfigEntity> eq =
            new LambdaQueryWrapper<FlinkNodeConfigEntity>()
                .eq(FlinkNodeConfigEntity::getProjectId, projectId)
                .eq(FlinkNodeConfigEntity::getNodeType, FlinkNodeTypeEnum.RESOURCE.getCode())
                .eq(FlinkNodeConfigEntity::getIsActive, 1);

        List<FlinkNodeConfigEntity> list = list(eq);

        return list;
    }
}
